package com.cnlive.logpare.page

import java.text.SimpleDateFormat
import java.text.SimpleDateFormat
import java.util.Locale
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import java.util.Date
import com.cnlive.logpare.util.IPSeekerExt
import scala.collection.mutable.HashMap
import java.net.URLDecoder
import com.cnlive.logpare.util.ParseLogUtil
import scala.util.control.Breaks._
import java.sql.DriverManager
import java.util.Properties
import java.sql.PreparedStatement
import java.sql.Connection
import com.cnlive.logpare.util.Constant
import java.net.URL
import java.util.UUID

object NewParsePageLog {
  val sdf=new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss",Locale.ENGLISH)
  val dateFormat=new SimpleDateFormat("yyyyMMddHHmmss",Locale.ENGLISH)
  val dateFormat3 =new SimpleDateFormat("yyyy-MM-dd", Locale.ENGLISH)
  def main(args: Array[String]): Unit = {
      var env=1
      if(args.length!=0){
        env=args(0).toInt
      }
      val conf = new SparkConf().setAppName("NewParsePageLog")
      if(env==1){
       conf.setMaster("local")
      }
       val sc = new SparkContext(conf)
       var date=dateFormat3.format(new Date())
       if(args.length!=0){
         date=args(1)
       }
       val files=env match{
         case 1 => "file:///F:/工作资料/page/*"
         case 2 => "hdfs://sxt/spark/test/input/logs/page/*"
         case 3 =>  "hdfs://sxt/nginx_logs/page/offline/"+date+"/*"
      }
      val qqwryPath=env match{
          case 2 =>Constant.qqwryLocalServer
          case 3 =>Constant.qqwryServer
          case _=>""
      }
      val zk_addr=env match{
        case 3=>"jdbc:phoenix:node001,node002,node003:2181"
        case _=>"jdbc:phoenix:node180,node181,node182:2181"
      }
      val person=sc.textFile(files).mapPartitions(x=>{
        var result = List[String]()
      val iPSeekerExt=env match {
          case 1 =>  IPSeekerExt.getInit()
          case 2 =>  IPSeekerExt.getInit(true,qqwryPath)
          case 3 =>  IPSeekerExt.getInit(true,qqwryPath)
      }
      val sessionMap=new HashMap[String, Int]()
      sessionMap.put("count", 0)
      val props=new Properties()
		  val conn = DriverManager.getConnection(zk_addr, props)
		  val nwesql="""
		    upsert into PAGELOG 
		    (NGINXTIME,HASHCODE,TM,UID,REF,URL,IP,DEVICE,BROWSER,PLAT,COUNTRY,CITY,LOCATION,OPERATOR,ISNEWUSER,FRMID) 
		     values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
		   """
		   val prest = conn.prepareStatement(nwesql) 
		    var count=0
        while (x.hasNext) {
           try{
              val lineText=x.next()
              count=count+1
              result= result.::(parLineText(iPSeekerExt,lineText,sessionMap,prest,conn))
              sessionMap.put("count",count)
           }catch {
             case t: Throwable => t.printStackTrace()
           }
        }
         conn.commit()
         conn.close()
         result.iterator
      }, true).count()
  }
 val ts=""" \"(.*?)\"""".r
  def parLineText(iPSeekerExt:IPSeekerExt,lineText:String,sessionMap:HashMap[String, Int],prest:PreparedStatement,con:Connection):String={
     var domain=""
     val mapRequest=new HashMap[String, String]()
     val ipandtime=ts.split(lineText).toList
     val urlandtype=ts.findAllIn(lineText).toList
     if(ipandtime.length==2&&urlandtype.length==3){
        val ipandtimeArr=ipandtime(0).split(" ")
        val timeIndx=ipandtimeArr.length-2
        val ip=if (ipandtimeArr(0).indexOf(",") != -1) ipandtimeArr(0).substring(0, ipandtimeArr(0).indexOf(",")) else ipandtimeArr(0)
        val time=ipandtimeArr(timeIndx).replace("[", "")
          mapRequest.put("nginxtime", time)//时间
        val d=iPSeekerExt.analyticIp(ip)
        val city=d.getCity
        val province=d.getProvince
        val country=d.getCountry
        val operator=d.getOperator
          mapRequest.put("ip", ip);//城市
          mapRequest.put("location", province);//省份
          mapRequest.put("city", city);//城市
          mapRequest.put("operator", operator+"")//运营商
       val urlandper=urlandtype(0).split(" ")
       if(urlandper.length==4){//解析url
          URLRequest(urlandper(2),mapRequest,sessionMap)
       }else{
          println("url解析有误"+lineText)
       }
       platParse(urlandtype(2),mapRequest)//解析操作系统
       if(!mapRequest.isEmpty){
         if(mapRequest.contains("nginxtime")&&
             mapRequest.contains("tm")&&mapRequest.contains("uid")
             &&mapRequest.contains("ip")){
            val nginxtime=dateFormat.format(sdf.parse(mapRequest.getOrElse("nginxtime", "20101010000000"))).toLong
            val tm=mapRequest.getOrElse("tm", "1000").toLong
            val uid=mapRequest.getOrElse("uid", "1000")
            val ref=mapRequest.get("ref") match {
              case  Some(u) =>new URL(u).getHost
              case  None=>""
            }
            val uri=mapRequest.get("uri")  match {
              case  Some(u) =>new URL(u).getHost
              case  None=>""
            }
            val appid=mapRequest.getOrElse("appid", "")//appid
            val ip=mapRequest.getOrElse("ip", "")//ip
            val device=mapRequest.getOrElse("device", "199").toInt//机型
            val browser=mapRequest.getOrElse("browser", "299").toInt//浏览器 
            val plat=mapRequest.getOrElse("browser", "299").toInt//操作系统
            val country=mapRequest.getOrElse("country", "99").toInt//国家
            val city=mapRequest.getOrElse("city", "")//城市
            val location=mapRequest.getOrElse("location", "")//省份
            val operator=mapRequest.getOrElse("operator", "99").toInt//运营商
            var isnewuser=mapRequest.getOrElse("isnewuser", "0").toInt//是否新用户
            val frmid=mapRequest.get("uri")  match {
              case  Some(u) =>{val d=new URL(u); 
                              val strUrlParam=d.getQuery;
                              if(strUrlParam!=null&&strUrlParam.nonEmpty){
                                parseUrlParam(strUrlParam,mapRequest)
                              }
                              mapRequest.getOrElse("frmid", "")
                }
              case  None=>""
            }
          //NGINXTIME,HASHCODE,TM,UID,REF,URI,IP,DEVICE,BROWSER,PLAT,COUNTRY,CITY,LOCATION,OPERATOR,ISNEWUSER,FRMID
          var index=1
          prest.setLong(index, nginxtime);index=index+1
          prest.setLong(index, lineText.hashCode());index=index+1
          prest.setLong(index, tm);index=index+1
          prest.setString(index, uid);index=index+1
          prest.setString(index, ref);index=index+1
          prest.setString(index, uri);index=index+1
          prest.setString(index, ip);index=index+1
          prest.setInt(index, device);index=index+1
          prest.setInt(index, browser);index=index+1
          prest.setInt(index, plat);index=index+1
          prest.setInt(index, country);index=index+1
          prest.setString(index, city);index=index+1
          prest.setString(index, location);index=index+1
          prest.setInt(index, operator);index=index+1
          prest.setInt(index, isnewuser);index=index+1
          prest.setString(index, frmid);index=index+1
          prest.execute()
         // if(sessionMap.getOrElse("count", 10)/100==0){
            con.commit()
          //}
         }else{
            println("nginxtime-tm,uid为空"+mapRequest)
         }
       }
     }else{
        println("数据格式不对"+lineText)
     }
     ""
  }
  val osList=List("nokia","sony","samsung", "htc", "sharp", "lenovo", "iphone", "ipad", "blackberry", "meizu",
            "android", "symbian", "ucweb", "opera", "cldc", "midp", "wap", "mobile", "htc", "motorola",
            "phone", "xiaomi", "oppo", "vivo", "honor", "huawei", "hisense", "zte", "playstation")
 //解析操作系统
  def platParse(text:String,mapRequest:HashMap[String, String]){
     var tmpbrowser=""
     try{
         tmpbrowser=URLDecoder.decode(text,"utf-8").toLowerCase(); 
     }catch {
        case t: Throwable => t.printStackTrace();tmpbrowser=text.toLowerCase() // TODO: handle error
    }
    if(tmpbrowser.indexOf("mozilla") != -1){
       tmpbrowser=tmpbrowser.substring(tmpbrowser.indexOf("mozilla")+11)
    }
     var flag=true
     breakable {
        for(os<-osList){//判断是移动端的是PC端的
          if(tmpbrowser.indexOf(os) != -1){
             flag=false
             break
           }
        }
      }
      if(flag){
          val  browserOs=ParseLogUtil.get_pc_browser(tmpbrowser)//PC浏览器类型
          val  plat=ParseLogUtil.get_pc_system(tmpbrowser)//PC系统类型
          mapRequest.put("browserOs",browserOs+"")
          mapRequest.put("plat",plat+"")
       }else{
          val  device=ParseLogUtil.get_decive(tmpbrowser)//手机型号
          val  browserOs=ParseLogUtil.get_mobile_browser(tmpbrowser)//手机浏览器类型
          val  plat=ParseLogUtil.get_mobile_system(tmpbrowser)//手机系统
          mapRequest.put("device",device+"")
          mapRequest.put("browserOs",browserOs+"")
          mapRequest.put("plat",plat+"")
       }
  }
  def parseUrlParam(strUrlParam:String,mapRequest:HashMap[String, String]){
     if(strUrlParam.length!=0){
     val arrSplit=strUrlParam.split("&")
     for(strSplit<-arrSplit){
       val arrSplitEqual=strSplit.split("[=]")
       if(arrSplitEqual.length>1){
         mapRequest.put(arrSplitEqual(0), strSplit.substring(strSplit.indexOf("=")+1));
       }
     }
   }
  }
 //url解析
  def URLRequest(url:String,mapRequest:HashMap[String, String],sessionMap:HashMap[String, Int]){
   val strUrlParam=TruncateUrlPage(URLDecoder.decode(url,"utf-8").toLowerCase())
   parseUrlParam(strUrlParam,mapRequest)
   val uid=mapRequest.getOrElse("uid", "")
   val tm=mapRequest.getOrElse("tm", "")
   if((!uid.equals(""))&&(!tm.equals(""))){
     val uidAndtm=uid.split("_")
     if(uidAndtm.length==2){
       if(uidAndtm(0).length()==13&&uidAndtm(0).matches("^[0-9]*$")){//uid中包含1494210652000_j1tsu24521
         if(uidAndtm(0).toLong==tm.toLong){
           if(sessionMap.contains(uid)){
            mapRequest.put("isnewuser", "0");   
           }else{
              sessionMap.put(uid, 1);
              mapRequest.put("isnewuser", "1");   
           }
         }else{
            mapRequest.put("isnewuser", "0"); 
         }
       }else if(uidAndtm(1).length()==13&&uidAndtm(1).matches("^[0-9]*$")){
           if(uidAndtm(1).toLong==tm.toLong&&uidAndtm(1).matches("^[0-9]*$")){
             if(sessionMap.contains(uid)){
               mapRequest.put("isnewuser", "0");   
             }else{
                sessionMap.put(uid, 1);
                mapRequest.put("isnewuser", "1");   
             }
           }else{
              mapRequest.put("isnewuser", "0"); 
           }
       }else{
          mapRequest.put("isnewuser", "0");
       }
     }
   }else{
     mapRequest.put("isnewuser", "0");
     println("url参数解析不合格！"+url)
   }
  }
  val url=new URL("""http://www.baidu.com""")
   //截取URL
  def TruncateUrlPage(strURL:String):String={
    val arrSplit=strURL.split("[?]")
    var strAllParam="";
    if(strURL.length()>1){
      if(arrSplit.length>1){
          if(arrSplit(1)!=null){
            val url2=new URL(url,strURL)
             strAllParam=url2.getQuery
          }
      }
    }
    strAllParam
  }
}